package com.lagou.rpcconsumer;

import com.lagou.service.JSONSerializer;
import com.lagou.service.RpcEncoder;
import com.lagou.service.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkMarshallingError;
import org.I0Itec.zkclient.serialize.ZkSerializer;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.*;

/**
 * @author zhangxuyuan
 * @version 1.0
 * @date 2020/11/29
 */
@Service
public class ProviderService {
    //将与服务提供者创建的连接保存进map，key为临时节点路径,value为Bootstrap
    Map<String, Bootstrap> bootstrapMap = new HashMap<>();
    //将与服务提供者通信的UserClientHandler保存进map，key为临时节点路径,value为UserClientHandler
    Map<String, UserClientHandler> handlerMap=new HashMap<>();
    //UserClientHandler列表
    List<UserClientHandler> handlerList = new ArrayList<>();
    //服务调用时的列表索引
    int i=0;

    @PostConstruct
    public void init()
    {   //初始化zk客户端
        ZkClient zkClient = new ZkClient("192.168.2.129:2181");
        //设置序列化
        zkClient.setZkSerializer(new ZkSerializer() {
            @Override public byte[] serialize(Object o) throws ZkMarshallingError {
                return o.toString().getBytes();
            }
            @Override public Object deserialize(byte[] bytes) throws ZkMarshallingError {
                return new String(bytes);
            }
        });
        //获取服务提供者，并创建连接
        List<String> nettyproviders = zkClient.getChildren("/nettyprovider");
        for(String path:nettyproviders)
        {
            Object o = zkClient.readData("/nettyprovider/"+path);
            Integer port = Integer.valueOf(o.toString());
            UserClientHandler handler=new UserClientHandler();
            Bootstrap bootStrap = this.createBootStrap(port, handler);
            //将创建的连接保存
            this.bootstrapMap.put(path,bootStrap);
            this.handlerMap.put(path,handler);
            this.handlerList.add(handler);
        }
        //注册监听器
        zkClient.subscribeChildChanges("/nettyprovider", new IZkChildListener() {
            @Override public void handleChildChange(String s, List<String> list) throws Exception {

                    Set<String> strings = ProviderService.this.bootstrapMap.keySet();
                    //剔除下线的服务
                    for(String childPath:strings)
                    {
                        if(!list.contains(childPath))
                        {
                            UserClientHandler handler = ProviderService.this.handlerMap.get(childPath);
                            ProviderService.this.handlerList.remove(handler);
                            ProviderService.this.handlerMap.remove(childPath);
                            ProviderService.this.bootstrapMap.remove(childPath);
                        }
                    }
                    //增加上线的服务
                    for(String childPath:list)
                    {
                        if(!ProviderService.this.bootstrapMap.containsKey(childPath))
                        {
                            Object o = zkClient.readData("/nettyprovider/"+childPath);
                            Integer port = Integer.valueOf(o.toString());
                            UserClientHandler handler=new UserClientHandler();
                            Bootstrap bootStrap = ProviderService.this.createBootStrap(port, handler);
                            ProviderService.this.bootstrapMap.put(childPath,bootStrap);
                            ProviderService.this.handlerMap.put(childPath,handler);
                            ProviderService.this.handlerList.add(handler);
                        }
                    }


            }
        });


    }

    public Bootstrap createBootStrap(Integer port,UserClientHandler handler)
    {
        //2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        //3)创建客户端的引导对象
        Bootstrap bootstrap =  new Bootstrap();
        //4)配置启动引导对象
        bootstrap.group(group)
            //设置通道为NIO
            .channel(NioSocketChannel.class)
            //设置请求协议为TCP
            .option(ChannelOption.TCP_NODELAY,true)
            //监听channel 并初始化
            .handler(new ChannelInitializer<SocketChannel>() {
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    //获取ChannelPipeline
                    ChannelPipeline pipeline = socketChannel.pipeline();
                    //设置编码
                    pipeline.addLast(new RpcEncoder(RpcRequest.class,new JSONSerializer()));
                    pipeline.addLast(new StringDecoder());
                    //添加自定义事件处理器
                    pipeline.addLast(handler);
                }
            });

        //5)连接服务端
        bootstrap.connect("127.0.0.1",port);
        return bootstrap;
    }

    public UserClientHandler setParam(RpcRequest rpcRequest)
    {   //每次获取通信handler时，轮询
        UserClientHandler handler = this.handlerList.get(this.i);
        handler.setParam(rpcRequest);
        i=(i+1)%this.handlerList.size();
        return handler;
    }

}
